package com.qyer.search.storm.ESClient;

import com.qyer.search.storm.exception.ESClientHostConfigException;
import com.qyer.search.storm.logging.InfoLogger;
import com.qyer.search.storm.util.ConfigUtils;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetAddress;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Created by tangzheng on 16/6/7.
 */
public class ESClientHolder {

  private InfoLogger dataLog;
  private volatile static ESClientHolder singleton;
  private TransportClient client;
  private AtomicBoolean _shutdown;

  //private Map<String, Object> config;

  private ESClientHolder(final Map<String, Object> config, String boltPrefix, String stormInfo) throws Exception{
    _shutdown = new AtomicBoolean(false);
    dataLog = new InfoLogger("/Data/search/ESClientHolder/" + stormInfo +"_cityService", stormInfo + "_data");
    Properties ESclientProp = ConfigUtils.configFromPrefix(config, ConfigUtils.INDEX_COMMON_PREFIX);
    String[] hosts = ESclientProp.getProperty("host", "esh1:9300,esh2:9300,esh3:9300").trim()
                                 .split(ConfigUtils.COMMON_SPLIT);
    //hosts = "esh3:9300,esh1:9300".split(ConfigUtils.COMMON_SPLIT);
    for(int i = 0; i < hosts.length; i++)
      dataLog.info("ESClient.hosts:{}", hosts[i]);
    String clusterName = ESclientProp.getProperty("cluster.name", "helloworld").trim();
    dataLog.info("ESClient.clusterName:{}", clusterName);
    Settings settings = Settings.settingsBuilder()
                                .put("cluster.name", clusterName)
                                .put("client.transport.sniff", true)
                                .build();
    client = TransportClient.builder()
                                            .settings(settings)
                                            .build();
    int size = hosts.length - 1;
    try {
      for (int i = size; i >= 0; i--) {
        String[] address = hosts[i].trim().split(":");
        client.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(address[0]),
                                                                  Integer.valueOf(address[1])));
      }
    }catch (Exception e){
      dataLog.info("ES client init failed,{}", e);
      throw new ESClientHostConfigException("ES client init failed", e);
    }
  }

  public static ESClientHolder getInstance(Map<String, Object> config, String boltPrefix, String stormInfo) throws Exception{
    if (singleton == null) {
      synchronized (ESClientHolder.class) {
        if (singleton == null) {
          singleton = new ESClientHolder(config, boltPrefix, stormInfo);
        }
      }
    }
    return singleton;
  }

  public TransportClient getClient() {
    return client;
  }

  public void setClient(TransportClient client) {
    this.client = client;
  }

  public void shutdown(){
    if(_shutdown.compareAndSet(false, true)) {
      try {
        if (client != null) {
          client.close();
          dataLog.info("es client shutdown ok ~");
        }
      } catch (Exception e) {
        dataLog.info("during shutdown,es_client connection shutdown failed.{}",e);
      }
    }
  }
}
